package com.liustar.mq.rocketmq;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * RocketMq监听wrapper，用于处理共通业务，并转发消息到业务监听中。
 *
 * @author Leo
 */
public class RocketMqMessageWrapper implements MessageListenerConcurrently {
    private MessageListener mqMessageListener;

    public MessageListener getMqMessageListener() {
        return mqMessageListener;
    }

    public void setMqMessageListener(MessageListener mqMessageListener) {
        this.mqMessageListener = mqMessageListener;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        //TODO 实现共通的业务
        Message message;
        if (list != null && list.size() > 0) {
            for (MessageExt msgx : list) {
                message = new Message();
                message.setTopic(msgx.getTopic());
                message.setTags(msgx.getTags());
                message.setKeys(msgx.getKeys());
                message.setBody(msgx.getBody());

                if (mqMessageListener.onMessage(message)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } else {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;

    }
}
